package org.duang.db.mongo;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.duang.db.Connection;
import org.duang.kit.ToolsKit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.mongodb.DB;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ReadPreference;
import com.mongodb.ServerAddress;

public class MongoConnection extends Connection {

	public static final String OPTIONS_FIELD = "options";
	
	
	private static Logger logger = LoggerFactory.getLogger(MongoConnection.class);
	
	protected MongoConnection mongodbConn;			// 连接对象
	private MongoClientOptions options;				// 自定义连接参数
	protected MongoClient client;					// 客户端对象
	protected DB db;								// db对象
	private List<ServerAddress> replicaSet;			// 数据集
    private List<MongoCredential> credentialsList;
   
    /**
     * 连接Mongodb数据库
     * @param optionMap		连接参数Map
     * @throws Exception
     */
    public void connect(Map<String, String> optionMap) throws Exception{
    	initConnect(optionMap);		//初始化连接参数
    }
    
    private void initConnect(Map<String, String> optionsMap) throws Exception{
    	
    	if( null == optionsMap || optionsMap.isEmpty()){
    		throw new NullPointerException("connection mongodb options map is null...");
    	}
    	
		this.host =  ToolsKit.isEmpty(optionsMap.get(HOST_FIELD)) ? "" : optionsMap.get(HOST_FIELD);    	
		this.port =  ToolsKit.isEmpty(optionsMap.get(PORT_FIELD)) ? 0 : Integer.parseInt(optionsMap.get(PORT_FIELD));
    	this.username = ToolsKit.isEmpty(optionsMap.get(USERNAME_FIELD)) ? "" : optionsMap.get(USERNAME_FIELD);
    	this.password = ToolsKit.isEmpty(optionsMap.get(PASSWORD_FIELD)) ? "" : optionsMap.get(PASSWORD_FIELD);
    	this.database = optionsMap.get(DATABASE_FIELD);
    	if(ToolsKit.isEmpty(database)){
    		throw new NullPointerException("database is null!");
    	}
    	String replics = optionsMap.get(REPLICASET_FIELD);
    	if(StringUtils.isNotBlank(replics)){
    		createReplics(replics);
    		options = MongoClientOptions.builder()
        								.connectionsPerHost(500)									// 最大连接数
        								.minConnectionsPerHost(10)
        								.heartbeatFrequency(10000)
        								.minHeartbeatFrequency(500)
        								.heartbeatConnectTimeout(15000)
        								.heartbeatSocketTimeout(20000)
        								.localThreshold(15)
        								.readPreference(ReadPreference.secondaryPreferred())		//优先在从节点中读数据,从节点有异常时再从主节点读数据
        								.connectTimeout(15000)
        								.maxWaitTime(120000)
        								.socketTimeout(0)
        								.threadsAllowedToBlockForConnectionMultiplier(50)			// 与connectionsPerHost相乘，变成一个线程变为可用的最大阻塞数
        								.build();
    	}
    	
    	initDasebase();  		//初始化数据库
    }
    
    // 如果有设置副本集地址,则创建,并重设host=null port=0
    private void createReplics(String replics) throws Exception{
    	String[] replicsArray = StringUtils.split(replics, ",");
    	if( null != replicsArray && replicsArray.length > 0 ){
    		replicaSet = new ArrayList<ServerAddress>(replicsArray.length);
    		for(String replic : replicsArray){
    			String[] replicItem = StringUtils.split(replic, ":");
    			if(null == replicItem || replicItem.length != 2){
    				throw new RuntimeException("replicsSet is null or length != 2 ");
    			}
    			int portItem = Integer.parseInt(replicItem[1]);
    			if(portItem <= 0) throw new RuntimeException("port not is number or <= 0!"); 
    			ServerAddress sa = new ServerAddress(replicItem[0], portItem);
    			logger.info("host: " + sa.getHost()+"           port: "+ sa.getPort());
    			replicaSet.add(sa);
    		}
    	}
    	if(null != replicaSet && !replicaSet.isEmpty()){
    		this.host = null;
    		this.port = 0;
    	}
    }
    
    @SuppressWarnings("deprecation")
	protected void initDasebase() {    	
    	try {
    		initAuth();// 初始化权限        	
			if(StringUtils.isNotBlank(host) && port > 0){
				ServerAddress sa = new ServerAddress(host, port);
				logger.info("host: " + sa.getHost()+"           port: "+ sa.getPort());
				if(null != options && null != credentialsList ){
					client = new MongoClient(sa, credentialsList, options);
				}else if(null != credentialsList){
					client = new MongoClient(sa, credentialsList);
				}else{
					client = new MongoClient(sa);
				}
			}else if(null != replicaSet){
				if(null != options && null != credentialsList ){
					client = new MongoClient(replicaSet, credentialsList, options);
				}else if(null == options && null != credentialsList ){					
					client = new MongoClient(replicaSet, credentialsList);
				}else{
					client = new MongoClient(replicaSet);
				}
			}
			
			if(null != client){
				db = client.getDB(database);		// client创建成功后就取出指定的库
				logger.info("Connected to mongodb successfully!");
			}else{
				throw new NullPointerException("can't connect mongodb database! crate client fail");
			}
		} catch (Exception e) {
			e.printStackTrace();
			logger.error(e.getMessage(), e);
			throw new RuntimeException(e.getMessage());
		}
	}

	protected void initAuth() {
		try{
			if(StringUtils.isNotBlank(username) && StringUtils.isNotBlank(password)){
				logger.info("username: " + username+"                password: " + password);
				MongoCredential credential = MongoCredential.createScramSha1Credential(username, database, password.toCharArray());
				if(credential == null){
					throw new RuntimeException("Can not connect to mongoDB. Failed to authenticate!");
				}else{
					credentialsList = Arrays.asList(credential);
				}
			}
		} catch (Exception e){
			logger.error(e.getMessage(), e);
			throw new RuntimeException(e.getMessage());
		}
	}

	
	/**
     * 关闭数据库所有连接
     */
    public void close(){
    	if(null != client){
    		db = null;
    		client.close();
    		mongodbConn.close();
    	}
    }

}
